package cas.ihep.test;

import com.google.common.io.Closer;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.input.PortableDataStream;

import java.io.BufferedReader;
import java.io.FileInputStream;
import java.io.InputStreamReader;
import java.util.ArrayList;
import java.util.HashMap;

public class SparkImport {
    public static void main(String args[])throws Exception{
        try(Closer closer=Closer.create()){
            final SparkConf conf = new SparkConf();
            final HashMap<String,String> mc=new HashMap<>();
            mc.put("target.directory",args[1]);
            JavaSparkContext jsc=closer.register(new JavaSparkContext(conf));
            FileInputStream lfis=closer.register(new FileInputStream(args[0]));
            InputStreamReader lireader=closer.register(new InputStreamReader(lfis));
            BufferedReader linput=closer.register(new BufferedReader(lireader));
            String line;
            ArrayList<String> inputfiles=new ArrayList<>();
            while((line=linput.readLine())!=null){
                if (line.length()>0){
                    inputfiles.add(line);
                }
            }
            JavaRDD<String> rdd0=jsc.parallelize(inputfiles,6);
            rdd0.map(new Function<String,String>(){
                @Override
                public String call(String s) throws Exception {
                    //Service4Hep s4h;

                    return null;
                }
            });
        }
    }
}
